package cn.jly.flink.connector;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 先将数据写到kafka中的topic（sink），再消费打印到控制台上（source）
 *
 * @author lanyangji
 * @create 2020-09-03 20:35
 */
public class KafkaConnectorApp {
    /**
     * id生成器
     */
    private static final AtomicLong ID_GENERATOR = new AtomicLong(1000);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStreamSource = env.fromElements(
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)).toString(),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)).toString(),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)).toString(),
                new JdbcConnectorApp.Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)).toString()
        );

        // flink producer
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                "hadoop101:9092", // broker list
                "flink-person-topic", // 目标topic
                new SimpleStringSchema()
        );

        // 0.10+ 版本的 Kafka 允许在将记录写入 Kafka 时附加记录的事件时间戳；
        // 此方法不适用于早期版本的 Kafka
        kafkaProducer.setWriteTimestampToKafka(true);

        // sink to kafka
        dataStreamSource.addSink(kafkaProducer);

        // 消费kafka中的数据，并打印到控制台
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "lanyangji-group");

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("flink-person-topic", new SimpleStringSchema(), props);
        // 尽可能从最早的记录开始消费
        kafkaConsumer.setStartFromEarliest();
//        kafkaConsumer.setStartFromLatest();       // 从最新的记录开始
//        kafkaConsumer.setStartFromTimestamp(...); // 从指定的时间开始（毫秒）
//        kafkaConsumer.setStartFromGroupOffsets(); // 默认的方法

        env.addSource(kafkaConsumer)
                .printToErr();

        env.execute("KafkaConnectorApp");
    }
}
